package com.gigi.searchcenterbackendmaster.job.cycle;

import cn.hutool.core.collection.CollUtil;
import com.gigi.searchcenterbackendmaster.esdao.UserEsDao;
import com.gigi.searchcenterbackendmaster.mapper.UserMapper;
import com.gigi.searchcenterbackendmaster.model.dto.UserEsDTO;
import com.gigi.searchcenterbackendmaster.model.entity.UserEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 增量同步用户数据到 ES
 *
 * @author Gigi
 * @date 2024/8/24 23:23
 */
@Slf4j
@Component
public class IncSyncUserToEs {

    @Resource
    private UserMapper userMapper;

    @Resource
    private UserEsDao userEsDao;

    /**
     * 每分钟执行一次
     */
    @Scheduled(fixedRate = 60 * 1000)
    public void run() {
        // 查询 5 分钟内更新的数据
        LocalDateTime fiveMinsAgo = LocalDateTime.now().minusMinutes(5);
        List<UserEntity> list = userMapper.listByUpdateTimeWithDelete(fiveMinsAgo);
        if (CollUtil.isEmpty(list)) {
            return;
        }
        // 同步到 ES
        List<UserEsDTO> userEsDTOS = list.stream().map(UserEsDTO::entityToDTO).collect(Collectors.toList());
        int total = list.size();
        final int pageSize = 500;
        log.info("Start to incrementally sync {} users to ES", total);
        for (int i = 0; i < total; i += pageSize) {
            int end = Math.min(i + pageSize, total);
            log.info("Incrementally syncing users from {} to {}", i, end);
            userEsDao.saveAll(userEsDTOS.subList(i, end));
        }
        log.info("Finish incrementally sync {} users to ES", total);
    }

}
